package com.k.activemq.openwire4;

import com.k.activemq.openwire4.domain.ConnData;
import com.k.activemq.openwire4.listener.KppMqListener;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Sub4 {
    public static void main(String[] args) {
        System.out.println("mq main start");
        // conn data
        String url = ConnData.url;
        String username = ConnData.username;
        String password = ConnData.password;
        String[] topics = ConnData.listenerTopics;
        Boolean pubSubDomain = ConnData.pubSubDomain;
        Boolean durableSubscription = ConnData.SubConfig.durableSubscription;

        // conn
        ActiveMQConnectionFactory connectionFactory = null;
        Connection connection = null;
        Session session = null;
        MessageConsumer consumer = null;
        try {
            // ConnectionFactory
            connectionFactory = new ActiveMQConnectionFactory(url);
            // 持久订阅需要设置
//            if(durableSubscription){
//                connectionFactory.setClientID("cid_20231106");
//            }
            // Connection
            connection = createConnectionWithStrategy(connectionFactory, username, password);//connectionFactory.createConnection(username, password);
            // 持久订阅需要设置,集群下 clientID 不能一样,否则会报错
            if(durableSubscription){
                connection.setClientID("cid_20240510");
            }
            connection.start();
            // Session
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // 多个目的地
            for (String topic : topics) {
                // 目的地
                // 消费者
                if(durableSubscription){
                    // 持久订阅,当订阅者下线，再上线，还可以接收到离线时期的消息。必须设置 clientID
                    // 只有 topic 支持持久订阅
                    consumer = session.createDurableSubscriber(session.createTopic(topic),"");
                }else{
                    Destination destination = pubSubDomain ? session.createTopic(topic) : session.createQueue(topic);
                    consumer = session.createConsumer(destination);
                }
                // 消费者 - 添加监听程序
                consumer.setMessageListener(new MessageListener() {
                    @Override
                    public void onMessage(Message message) {
                        try {
                            // 消费 message
                            System.out.println("消费者接收到消息："+message);
                            // 如果 replay=null 则该消息未设置回应地址
                            System.out.println("消费者接收：replay:"+message.getJMSReplyTo());
                            // 消费回执，通知 ActiveMQ 消息已被消费,创建 session时指定自动ack这里就不用再ack了
//                            message.acknowledge();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }
        System.out.println("mq main finish");
    }

    private static Connection createConnectionWithStrategy(ConnectionFactory connectionFactory, String username, String password) {
        try {
            return connectionFactory.createConnection(username, password);
        } catch (Exception e) {
            System.out.println("连接失败");
            try {
                Thread.sleep(5000);
                createConnectionWithStrategy(connectionFactory, username, password);
            } catch (InterruptedException interruptedException) {
                interruptedException.printStackTrace();
            }
        }
        return null;
    }
}
